/*
 * @version: 1.0
 * @Author: zhl
 * @Date: 2021-09-22 10:43:39
 * @LastEditors: zhl
 * @LastEditTime: 2022-02-08 13:27:50
 * @Description: 文件实现描述
 */
#include "HandleThreadManager.h"
#include "CLogger.h"
#include "app.h"
#include "IModule.h"
#include "HPLCArchivesManager.h"
#include <memory>
#include <string>
#include <iostream>

typedef void  (*TFUNCTION_WAIT)(std::string);

#define LOWEST_PRIORITY_LEVEL  15
#define RECV_BUF_MAX_SIZE 8192

HandleThreadManager::HandleThreadManager(std::shared_ptr<CConfigManager> cm, std::shared_ptr<CMosquittoManager> mm, 
    std::string addr, int comm_type){
    c_is_run = true;
    c_cm = cm;
    c_mm = mm;
    c_recv_data = (uint8_t *)malloc(RECV_BUF_MAX_SIZE);
    c_sp_ptr = nullptr;
    c_hplc_ptr = nullptr;

    c_comm_type = comm_type;
    if (PORT_TYPE_SERIAL == comm_type) {
        init_serial(addr);
    }
    else if (PORT_TYPE_HPLC == comm_type)
    {
        init_hplc();
    }
    
}

HandleThreadManager::~HandleThreadManager(){
    if(NULL != c_recv_data){
        free(c_recv_data);
        c_recv_data = NULL;
    }
}

void HandleThreadManager::init_serial(std::string &addr){
    c_sp_ptr = std::make_shared<SerialProxy>(c_cm, std::move(addr));
    c_sp_ptr->connect(mosquitto_addr, mosquitto_port, NULL, NULL, 10);
    auto serial = c_cm->get_serial_info(c_sp_ptr->c_if_addr);
    SERIAL_PARAMS params;
    if(nullptr == serial){
        params.baudRate = 9600;
        params.byteSize = 8;
        params.parity   = "none";
        params.stopBits = 1; 
    } else {
        params.baudRate = serial->baud_rate;
        params.byteSize = serial->data_size;
        params.parity   = serial->parity;
        params.stopBits = serial->stop_bit; 
    }
    c_sp_ptr->set_serial_params(std::move(params));
}

void HandleThreadManager::stop(){
    c_is_run = false;
    pthread_cancel(c_pthread);
    clear_task_items();
}

int HandleThreadManager::execute(){
    c_is_run = true;
    int ret = pthread_create(&c_pthread, NULL, handle_thread, this);
    return ret;
}
//
int HandleThreadManager::run(std::shared_ptr<TASK_PARAMS> dev){
    auto v_dev = c_cm->get_terminal_device(dev->dev_id);
    if(nullptr == v_dev){
        CLogger::getInstance()->LOG_W("该设备不存在");
        dev->items.pop_front();
        return -1;
    }
    auto rule = c_cm->get_protocol_rules()->find(v_dev->protocol_id);

    if(rule ==  c_cm->get_protocol_rules()->end()){
        CLogger::getInstance()->LOG_W("不支持该协议的处理");
        dev->items.pop_front();
        return -1;
    }else{
        //针对645、modbus临时处理
        std::shared_ptr<TASK_PARAMS> task_params = nullptr;
        if((1 == rule->second.protocol_type) || (2 == rule->second.protocol_type)){
            task_params = std::make_shared<TASK_PARAMS>();
            task_params->dev_id  = dev->dev_id;
            task_params->task_id = dev->task_id;
            task_params->priority  = dev->priority;
            task_params->req_id  = dev->req_id;
            task_params->dev  = dev->dev;
            task_params->topic = dev->topic;
            task_params->type  = dev->type;
            task_params->cmd_source  = dev->cmd_source;
            task_params->items.clear();
            task_params->items.push_back(dev->items.front());
            dev->items.pop_front();
        } else {
            task_params = dev;
        }
        TFUNCTION_RUN func_run = (TFUNCTION_RUN)IModule::getInstance().get_func(std::to_string(rule->second.protocol_type), "run");
        if(NULL == func_run){
            dev->items.pop_front();
            return -1;
        }
        if (PORT_TYPE_HPLC == c_comm_type) {
            func_run(c_cm, c_hplc_ptr, c_mm, IModule::getInstance(), task_params);
        }
        else{
            func_run(c_cm, c_sp_ptr, c_mm, IModule::getInstance(), task_params);
        }
        
    }
    return 0;
}
void HandleThreadManager::wait(int ms){

}
void *HandleThreadManager::response_callback(void *arg){
    return NULL;
}

void *HandleThreadManager::handle_thread(void *arg){
    pthread_detach(pthread_self());
    if(NULL == arg){
        return NULL;
    }
    HandleThreadManager *htm = (HandleThreadManager *)arg;
    while(htm->c_is_run){
        htm->c_notify.lock();
        while(!htm->is_exists_task_item()){
            htm->c_notify.wait();
        }
        //先执行命令
        std::shared_ptr<TASK_PARAMS> cmd_dev = htm->get_dev_task_items(htm->c_cmd_devs);
        if(nullptr == cmd_dev){
            std::shared_ptr<TASK_PARAMS> dev = htm->get_dev_task_items(htm->c_devs);
            htm->c_notify.unlock();
            if(nullptr != dev){
                while(!dev->items.empty()){
                    htm->c_notify.lock();
                    cmd_dev = htm->get_dev_task_items(htm->c_cmd_devs);
                    htm->c_notify.unlock();
                    if(nullptr == cmd_dev){
                        if(-1 == htm->run(dev)){
                            //sleep(1);
                        }
                    } else {
                        if(-1 == htm->run(cmd_dev)){
                            //sleep(1);
                        }
                    }
                }
            }
        } else {
            htm->c_notify.unlock();
            if(-1 == htm->run(cmd_dev)){
                //sleep(1);
            }
        }

    }
    return NULL;
}

bool HandleThreadManager::is_exists_task_item(){
    for(int i = 0; i < LOWEST_PRIORITY_LEVEL; i++){
        auto task_dev_list = c_devs.find(i);
        if(task_dev_list != c_devs.end()){
            if(!task_dev_list->second.empty()){
                return true;
            }            
        }
    }
    for(int i = 0; i < LOWEST_PRIORITY_LEVEL; i++){
        auto task_dev_list = c_cmd_devs.find(i);
        if(task_dev_list != c_cmd_devs.end()){
            if(!task_dev_list->second.empty()){
                return true;
            } 
        }
    }
    return false;
}

std::shared_ptr<TASK_PARAMS> HandleThreadManager::get_dev_task_items(std::unordered_map<int64_t, std::list<TASK_PARAMS>> &devs){
    for(int i = 0; i < LOWEST_PRIORITY_LEVEL; i++){
        auto task_dev_list = devs.find(i);
        if(task_dev_list != devs.end()){
            if(!task_dev_list->second.empty()){
                auto dev = task_dev_list->second.front();
                task_dev_list->second.pop_front();
                return std::make_shared<TASK_PARAMS>(dev);
            }
        }
    }
    return nullptr;
}
void HandleThreadManager::push_task_item(TASK_PARAMS &dev, std::unordered_map<int64_t, std::list<TASK_PARAMS>> &devs){
    if(dev.items.empty()){
        return ;
    }
    auto task_dev_list = devs.find(dev.priority);
    if(task_dev_list == devs.end()){
        std::list<TASK_PARAMS> tdl;
        tdl.push_back(dev);
        devs.insert(std::make_pair(dev.priority, tdl));
    }else{
        task_dev_list->second.push_back(dev);
    }
}

void HandleThreadManager::add_task_items(TASK_PARAMS &dev){
    c_notify.lock();
    push_task_item(dev, c_devs);
    c_notify.signal();
    c_notify.unlock();
}
//
void HandleThreadManager::add_cmd_task_items(TASK_PARAMS &dev){
    c_notify.lock();
    push_task_item(dev, c_cmd_devs);
    c_notify.signal();
    c_notify.unlock();
}
void HandleThreadManager::clear_task_items(){
    c_notify.lock();
    c_devs.clear();
    c_cmd_devs.clear();
    c_notify.unlock();
}

void HandleThreadManager::init_hplc()
{
    c_hplc_ptr = std::make_shared<HPLCProxy>(1);
    if (nullptr == c_hplc_ptr) {
        CLogger::getInstance()->LOG_E("init_hplc error,c_hplc_ptr is null.");
        return;
    }

    c_hplc_ptr->connect(mosquitto_addr, mosquitto_port,  NULL, NULL, 10);

    std::shared_ptr<HPLCArchivesManager> hplc_archives = std::make_shared<HPLCArchivesManager>(c_cm, c_hplc_ptr);
    hplc_archives->ArchivesRegister();
}